package cn.itcast.flink.base;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;
import java.util.UUID;


/**
 * Author itcast
 * Date 2021/7/27 10:53
 * 每隔1秒随机生成一条订单信息(订单ID、用户ID、订单金额、时间戳)
 * 要求:
 * - 随机生成订单ID(UUID)
 * - 随机生成用户ID(0-2)
 * - 随机生成订单金额(0-100)
 * - 时间戳为当前系统时间
 */
public class CustomSource {
    public static void main(String[] args) throws Exception {
        //todo 1.获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //todo 2.获取用户自定义数据源
        //todo 2.1.生成用户自定义类实现业务逻辑
        DataStreamSource<Order> source = env.addSource(new SourceFunction<Order>() {
            boolean flag = true;
            Random rn = new Random();
            //todo 2.2.模拟订单信息
            /**
             * 实现每一条订单的数据
             * @param ctx 收集当前生成的每一条数据
             * @throws Exception
             */
            @Override
            public void run(SourceContext<Order> ctx) throws Exception {
                while (flag) {
                    String oid = UUID.randomUUID().toString();
                    int uid = rn.nextInt(3);
                    int money = rn.nextInt(101);
                    long currentTime = System.currentTimeMillis();
                    ctx.collect(new Order(oid, uid, money, currentTime));
                    Thread.sleep(1000L);
                }
            }

            /**
             * 取消当前的数据源的生成任务
             */
            @Override
            public void cancel() {
                flag = false;
            }
        });


        //todo 3.打印输出
        source.print();
        //todo 4.执行流环境
        env.execute();
    }

    //todo 2.3.生成一个订单 JavaBean 对象
    //定义一个Order对象
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Order{
        //订单ID
        private String oid;
        //用户ID
        private int uid;
        //订单金额
        private int money;
        //获取时间戳
        private Long currentTime;
    }
}
